/***
* ==++==
*
* Copyright (c) Microsoft Corporation.  All rights reserved.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* pid.h
*
* C++ Actors: PID: one of the main user-facing C++ Actors APIs.
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/

#pragma once

#include "actor_public.h"
#include "msg_ptr.h"
#include "procinfo.h"
#include "xxagents.h"
#include <safeint.h>

namespace actors
{
    /// <summary>
    /// PID (short for "Process IDentifier") is the only means to communicate with
    /// a process. PIDs can be serialized and sent across in a message and used from
    /// more than one client.
    /// </summary>
    class PID
    {
    public:
        /// <summary>
        /// Constructor
        /// </summary>
        _DISPIMP PID ();
        /// <summary>
        /// Constructor
        /// </summary>
        /// <param name="id">The concrete GUID to use for the PID</param>
        /// <param name="uri">The uri to connect to when spawning the process</param>
        _DISPIMP PID (const msg_guid &id, const std::string &uri = "");

        /// <summary>
        /// A group of actor identifiers
        /// </summary>
        /// <remarks>Groups of PIDs have order and need to be vectors rather than sets.</remarks>
        // TODO: consider how to avoid duplication within a group of PIDs. TFS: # 336219
        typedef std::vector<PID> Group;

        /// <summary>
        /// Invalidate will render a PID instance unusable for sending messages
        /// </summary>
        _DISPIMP PID & invalidate();

        /// <summary>
        /// Check whether the actor identifier can be used for sending messages
        /// </summary>
        _DISPIMP bool is_valid() const; 

        /// <summary>
        /// Get the GUID of the actor identifier
        /// </summary>
        inline const msg_guid &id() const { return m_actor_info ? m_actor_info->guid : msg_guid::null; }

        /// <summary>
        /// Get the URI of the actor.
        /// </summary>
        _DISPIMP std::string uri() const;

        /// <summary>
        /// Send the actor represented by the identifier a PING request and
        /// return true if it is available.
        /// </summary>
        _DISPIMP bool is_available() const;

        /// <summary>
        /// Is the actor represented by the identifier local or remote?
        /// </summary>
        _DISPIMP bool is_local() const;

        /// <summary>
        /// Represent the actor identifier as a string.
        /// </summary>
        _DISPIMP operator std::string() const { return to_string(); }

        _DISPIMP void trace_messages(bool trace) { m_trace_msgs = trace; }

        //
        // It is convenient to be able to compare PIDs so that they
        // can be used as keys in tables, etc.
        //
        // TODO: identify the bug that causes sets of PIDs to not work... TFS: # 336218
        _DISPIMP bool operator == (const PID &other) const; 
        _DISPIMP bool operator != (const PID &other) const; 
        _DISPIMP bool operator  < (const PID &other) const; 
        _DISPIMP bool operator  > (const PID &other) const; 

    #pragma region send()
        /// <summary>
        /// Send is the originating construct for all simplex point-to-point messages.
        /// </summary>
        /// <param name="data">A message payload</param>
        template<typename Payload>
        bool send(const Payload &data) const
        {
            if ( !is_valid() || m_actor_info->chan == nullptr ) return false;

            bool result = false;

            if ( m_actor_info->is_local() )
            {
                local_msg<Payload> *msg = m_actor_info->_allocate_local<Payload>(typeid(Payload), m_trace_msgs);
                msg->data = data;

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) free(msg);
            }
            else
            {
                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(Payload).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(data, *serializer) && serializer->flush();
            }

            return result;
        }

        /// <summary>
        /// Send a message to a group of actors.
        /// </summary>
        /// <param name="group">A group of actor identifiers</param>
        /// <param name="data">A message payload</param>
        template<typename Payload>
        static std::vector<bool> send(const Group &group, const Payload &data)
        {
            std::vector<bool> result;

            size_t length = ::calculate_size(data);
            byte *buffer = new byte[msl::utilities::SafeInt<size_t>(length)];

            array_serializer serializer(length, buffer);

            // Naive implementation, potentially sending more than once to each host

            if (::serialize(data, serializer))
            {
                for (auto pos = group.begin(); pos != group.end(); pos++)
                {
                    if ( !pos->is_valid() ) { result.push_back(false); continue; }

                    if ( pos->m_actor_info->is_local() )
                    {
                        local_msg<Payload> *msg = pos->m_actor_info->_allocate_local<Payload>(typeid(Payload), pos->m_trace_msgs);
                        msg->data = data;

                        bool res = pos->m_actor_info->chan->dispatch((dispatch_msg *)msg);

                        if (!res) free(msg);

                        result.push_back(res);
                    }
                    else
                    {
                        std::auto_ptr<Serializer> auto_ptr_serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(Payload).name(), pos->m_actor_info, pos->m_trace_msgs));

                        if ( auto_ptr_serializer.get() != nullptr )
                            result.push_back(auto_ptr_serializer->write(length, buffer) && auto_ptr_serializer->flush());
                    }
                }
            }
            else
            {
                for (size_t i = 0; i < group.size(); i++) result.push_back(false);
            }

            delete [] buffer;

            return result;
        }

        /// <summary>
        /// Send a pair of data as a single message.
        /// </summary>
        /// <param name="data0">A first message payload</param>
        /// <param name="data1">A second message payload</param>
        template<typename Payload0, typename Payload1>
        bool send(const Payload0 &data0, const Payload1 &data1) const
        {
            if ( !is_valid() || m_actor_info->chan == nullptr ) return false;

            bool result = false;
        
            typedef std::tuple<Payload0,Payload1> _tuple;

            if ( m_actor_info->is_local() )
            {
                auto msg = m_actor_info->_allocate_local<_tuple>(typeid(_tuple), m_trace_msgs);
                msg->data = std::tie(data0, data1);

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) 
                    free(msg);
            }
            else
            {
                _tuple data = std::tie(data0, data1);

                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(_tuple).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(data, *serializer) && serializer->flush();
            }

            return result;
        }

        /// <summary>
        /// Send a pair of data as a single message to a group of actors.
        /// </summary>
        /// <param name="group">A group of actor identifiers</param>
        /// <param name="data0">A first message payload</param>
        /// <param name="data1">A second message payload</param>
        template<typename Payload0, typename Payload1>
        static std::vector<bool> send(const PID::Group &group, const Payload0 &data0, const Payload1 &data1)
        {
            std::vector<bool> result;

            typedef std::tuple<Payload0,Payload1> _tuple;

            _tuple data = std::tie(data0, data1);

            size_t length = ::calculate_size(data);
            void *buffer = new char[msl::utilities::SafeInt<size_t>(length)];

            // Naive implementation, potentially sending more than once to each host
            array_serializer serializer(length, buffer);
            if (::serialize(data, serializer))
            {
                for (auto pos = group.begin(); pos != group.end(); pos++)
                {
                    if ( !pos->is_valid() ) { result.push_back(false); continue; }

                    if ( pos->m_actor_info->is_local() )
                    {
                        local_msg<_tuple> *msg = pos->m_actor_info->_allocate_local<_tuple>(typeid(_tuple), pos->m_trace_msgs);
                        msg->data = data;

                        bool res = pos->m_actor_info->chan->dispatch((dispatch_msg *)msg);

                        if (!res) free(msg);

                        result.push_back(res);
                    }
                    else
                    {
                        std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(_tuple).name(), pos->m_actor_info, pos->m_trace_msgs));

                        if ( serializer.get() != nullptr )
                            result.push_back(serializer->write(length, buffer) && serializer->flush());
                    }
                }
            }
            else
            {
                for (size_t i = 0; i < group.size(); i++) result.push_back(false);
            }

            delete buffer;

            return result;
        }

        /// <summary>
        /// Send a triple of data as a single message.
        /// </summary>
        /// <param name="data0">A first message payload</param>
        /// <param name="data1">A second message payload</param>
        /// <param name="data2">A third message payload</param>
        template<typename Payload0, typename Payload1, typename Payload2>
        bool send(const Payload0 &data0, const Payload1 &data1, const Payload2 &data2) const
        {
            if ( !is_valid() || !m_actor_info->chan ) return false;

            bool result = false;
        
            typedef std::tuple<Payload0,Payload1,Payload2> _tuple;

            if ( m_actor_info->is_local() )
            {
                auto msg = m_actor_info->_allocate_local<_tuple>(typeid(_tuple), m_trace_msgs);
                msg->data = std::tie(data0, data1, data2);

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) 
                    free(msg);
            }
            else
            {
                _tuple data = std::tie(data0, data1, data2);

                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(_tuple).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(data, *serializer) && serializer->flush();
            }
            return result;
        }

        /// <summary>
        /// Send a triple of data as a single message to a group of actors.
        /// </summary>
        /// <param name="group">A group of actor identifiers</param>
        /// <param name="data0">A first message payload</param>
        /// <param name="data1">A second message payload</param>
        /// <param name="data2">A third message payload</param>
        template<typename Payload0, typename Payload1, typename Payload2>
        static std::vector<bool> send(const PID::Group &group, const Payload0 &data0, const Payload1 &data1, const Payload2 &data2)
        {
            std::vector<bool> result;

            typedef std::tuple<Payload0,Payload1,Payload2> _tuple;

            _tuple data = std::tie(data0, data1, data2);

            size_t length = ::calculate_size(data), len = length;
            void *buffer = new char[msl::utilities::SafeInt<size_t>(length)], *buf = buffer;

            // Naive implementation, potentially sending more than once to each host
            array_serializer serializer(len, buf);
            if (::serialize(data, serializer))
            {
                for (auto pos = group.begin(); pos != group.end(); pos++)
                {
                    if ( !pos->is_valid() ) { result.push_back(false); continue; }

                    if ( pos->m_actor_info->is_local() )
                    {
                        local_msg<_tuple> *msg = pos->m_actor_info->_allocate_local<_tuple>(typeid(_tuple), pos->m_trace_msgs);
                        msg->data = data;

                        bool res = pos->m_actor_info->chan->dispatch((dispatch_msg *)msg);

                        if (!res) free(msg);

                        result.push_back(res);
                    }
                    else
                    {
                        std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(_tuple).name(), pos->m_actor_info, pos->m_trace_msgs));

                        if ( serializer.get() != nullptr )
                            result.push_back(serializer->write(length, buffer) && serializer->flush());
                    }
                }
            }
            else
            {
                for (size_t i = 0; i < group.size(); i++) result.push_back(false);
            }

            delete buffer;

            return result;
        }

        /// <summary>
        /// Send a quadruple of data as a single message.
        /// </summary>
        /// <param name="data0">A first message payload</param>
        /// <param name="data1">A second message payload</param>
        /// <param name="data2">A third message payload</param>
        /// <param name="data3">A fourth message payload</param>
        template<typename Payload0, typename Payload1, typename Payload2, typename Payload3>
        bool send(const Payload0 &data0, const Payload1 &data1, const Payload2 &data2, const Payload3 &data3) const
        {
            if ( !is_valid() || !m_actor_info->chan ) return false;

            bool result = false;
        
            typedef std::tuple<Payload0,Payload1,Payload2,Payload3> _tuple;

            if ( m_actor_info->is_local() )
            {
                auto msg = m_actor_info->_allocate_local<_tuple>(typeid(_tuple), m_trace_msgs);
                msg->data = std::tie(data0, data1, data2, data3);

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) 
                    free(msg);
            }
            else
            {
                _tuple data = std::tie(data0, data1, data2, data3);

                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(_tuple).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(data, *serializer) && serializer->flush();
            }

            return result;
        }

        /// <summary>
        /// Send a quadruple of data as a single message to a group of actors.
        /// </summary>
        /// <param name="group">A group of actor identifiers</param>
        /// <param name="data0">A first message payload</param>
        /// <param name="data1">A second message payload</param>
        /// <param name="data2">A third message payload</param>
        /// <param name="data3">A fourth message payload</param>
        template<typename Payload0, typename Payload1, typename Payload2, typename Payload3>
        static std::vector<bool> send(const PID::Group &group, const Payload0 &data0, const Payload1 &data1, const Payload2 &data2, const Payload3 &data3)
        {
            std::vector<bool> result;

            typedef std::tuple<Payload0,Payload1,Payload2,Payload3> _tuple;

            _tuple data = std::tie(data0, data1, data2, data3);

            size_t length = ::calculate_size(data), len = length;
            void *buffer = new char[msl::utilities::SafeInt<size_t>(length)], *buf = buffer;

            // Naive implementation, potentially sending more than once to each host
            array_serializer serializer(len, buf);
            if (::serialize(data, serializer))
            {
                for (auto pos = group.begin(); pos != group.end(); pos++)
                {
                    if ( !pos->is_valid() ) { result.push_back(false); continue; }

                    if ( pos->m_actor_info->is_local() )
                    {
                        local_msg<_tuple> *msg = pos->m_actor_info->_allocate_local<_tuple>(typeid(_tuple), pos->m_trace_msgs);
                        msg->data = data;

                        bool res = pos->m_actor_info->chan->dispatch((dispatch_msg *)msg);

                        if (!res) free(msg);

                        result.push_back(res);
                    }
                    else
                    {
                        std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(_tuple).name(), pos->m_actor_info, pos->m_trace_msgs));

                        if ( serializer.get() != nullptr )
                            result.push_back(serializer->write(length, buffer) && serializer->flush());
                    }
                }
            }
            else
            {
                for (size_t i = 0; i < group.size(); i++) result.push_back(false);
            }

            delete buffer;

            return result;
        }
    #pragma endregion

    #pragma region request()
        /// <summary>
        /// Request is used for duplex messages, where a reply is correlated
        /// with each request message. This is essentially a send/receive
        /// combo, and is (currently) a blocking construct.
        /// </summary>
        /// <typeparam name="ReqT">
        /// The outgoing payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The returning payload type 
        /// </typeparam>
        /// <param name="input">
        /// A reference to the outgoing data.
        /// </param>
        /// <param name="output">
        /// A reference to a variable where the result will be placed.
        /// </param>
        /// <returns>
        /// The result of the serialization and send operation.
        /// </returns>
        template<typename ReqT, typename RepT>
        bool request(const ReqT &input, RepT &output) const
        {
            typedef reqrep_msg<ReqT,RepT> msg_type;

            if ( !is_valid() || !m_actor_info->chan ) return false;

            bool result = false;
        
            Concurrency::overwrite_buffer<RepT> reply;

            msg_type rrmsg;
            rrmsg.data = input;
            rrmsg.reply_to = PID::forward(&reply, true);

            if ( m_actor_info->is_local() )
            {
                local_msg<msg_type> *msg = m_actor_info->_allocate_local<msg_type>(typeid(msg_type), m_trace_msgs);
                msg->data = rrmsg;

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) free(msg);
            }
            else
            {
                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(msg_type).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(rrmsg, *serializer) && serializer->flush();
            }

            if (result)
                output = Concurrency::receive<RepT>((Concurrency::ISource<RepT> *)&reply);

            rrmsg.reply_to.invalidate();

            return result;
        }

        /// <summary>
        /// Request is used for duplex messages, where a reply is correlated
        /// with each request message. This is essentially a send/receive
        /// combo, and is (currently) a blocking construct.
        /// </summary>
        /// <typeparam name="ReqT_1">
        /// The first outgoing payload type 
        /// </typeparam>
        /// <typeparam name="ReqT_2">
        /// The second outgoing payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The returning payload type 
        /// </typeparam>
        /// <param name="input1">
        /// A reference to the first outgoing data.
        /// </param>
        /// <param name="input2">
        /// A reference to the second outgoing data.
        /// </param>
        /// <param name="output">
        /// A reference to a variable where the result will be placed.
        /// </param>
        /// <returns>
        /// The result of the serialization and send operation.
        /// </returns>
        template<typename ReqT_1, typename ReqT_2, typename RepT>
        bool request(const ReqT_1 &input1, const ReqT_2 &input2, RepT &output) const
        {
            typedef reqrep_msg<std::tuple<ReqT_1,ReqT_2>,RepT> msg_type;

            if ( !is_valid() || !m_actor_info->chan ) return false;

            bool result = false;
        
            Concurrency::overwrite_buffer<RepT> reply;

            msg_type rrmsg;
            rrmsg.data = std::make_tuple(input1,input2);
            rrmsg.reply_to = PID::forward(&reply, true);

            if ( m_actor_info->is_local() )
            {
                local_msg<msg_type> *msg = m_actor_info->_allocate_local<msg_type>(typeid(msg_type), m_trace_msgs);
                msg->data = rrmsg;

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) free(msg);
            }
            else
            {
                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(msg_type).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(rrmsg, *serializer) && serializer->flush();
            }

            if (result)
                output = Concurrency::receive<RepT>((Concurrency::ISource<RepT> *)&reply);

            rrmsg.reply_to.invalidate();

            return result;
        }

        /// <summary>
        /// Request is used for duplex messages, where a reply is correlated
        /// with each request message. This is essentially a send/receive
        /// combo, and is (currently) a blocking construct.
        /// </summary>
        /// <typeparam name="ReqT_1">
        /// The first outgoing payload type 
        /// </typeparam>
        /// <typeparam name="ReqT_2">
        /// The second outgoing payload type 
        /// </typeparam>
        /// <typeparam name="ReqT_3">
        /// The third outgoing payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The returning payload type 
        /// </typeparam>
        /// <param name="input1">
        /// A reference to the first outgoing data.
        /// </param>
        /// <param name="input2">
        /// A reference to the second outgoing data.
        /// </param>
        /// <param name="input3">
        /// A reference to the third outgoing data.
        /// </param>
        /// <param name="output">
        /// A reference to a variable where the result will be placed.
        /// </param>
        /// <returns>
        /// The result of the serialization and send operation.
        /// </returns>
        template<typename ReqT_1, typename ReqT_2, typename ReqT_3, typename RepT>
        bool request(const ReqT_1 &input1, const ReqT_2 &input2, const ReqT_3 &input3, RepT &output) const
        {
            typedef reqrep_msg<std::tuple<ReqT_1,ReqT_2,ReqT_3>,RepT> msg_type;

            if ( !is_valid() || !m_actor_info->chan ) return false;

            bool result = false;
        
            Concurrency::overwrite_buffer<RepT> reply;

            msg_type rrmsg;
            rrmsg.data = std::make_tuple(input1,input2,input3);
            rrmsg.reply_to = PID::forward(&reply, true);

            if ( m_actor_info->is_local() )
            {
                local_msg<msg_type> *msg = m_actor_info->_allocate_local<msg_type>(typeid(msg_type), m_trace_msgs);
                msg->data = rrmsg;

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) free(msg);
            }
            else
            {
                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(msg_type).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(rrmsg, *serializer) && serializer->flush();
            }

            if (result)
                output = Concurrency::receive<RepT>((Concurrency::ISource<RepT> *)&reply);

            rrmsg.reply_to.invalidate();

            return result;
        }

        /// <summary>
        /// Request is used for duplex messages, where a reply is correlated
        /// with each request message. This is essentially a send/receive
        /// combo, and is (currently) a blocking construct.
        /// </summary>
        /// <typeparam name="ReqT_1">
        /// The first outgoing payload type 
        /// </typeparam>
        /// <typeparam name="ReqT_2">
        /// The second outgoing payload type 
        /// </typeparam>
        /// <typeparam name="ReqT_3">
        /// The third outgoing payload type 
        /// </typeparam>
        /// <typeparam name="ReqT_4">
        /// The fourth outgoing payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The returning payload type 
        /// </typeparam>
        /// <param name="input1">
        /// A reference to the first outgoing data.
        /// </param>
        /// <param name="input2">
        /// A reference to the second outgoing data.
        /// </param>
        /// <param name="input3">
        /// A reference to the third outgoing data.
        /// </param>
        /// <param name="input4">
        /// A reference to the fourth outgoing data.
        /// </param>
        /// <param name="output">
        /// A reference to a variable where the result will be placed.
        /// </param>
        /// <returns>
        /// The result of the serialization and send operation.
        /// </returns>
        template<typename ReqT_1, typename ReqT_2, typename ReqT_3, typename ReqT_4, typename RepT>
        bool request(const ReqT_1 &input1, const ReqT_2 &input2, const ReqT_3 &input3, const ReqT_4 &input4, RepT &output) const
        {
            typedef reqrep_msg<std::tuple<ReqT_1,ReqT_2,ReqT_3,ReqT_4>,RepT> msg_type;

            if ( !is_valid() || !m_actor_info->chan ) return false;

            bool result = false;
        
            Concurrency::overwrite_buffer<RepT> reply;

            msg_type rrmsg;
            rrmsg.data = std::make_tuple(input1,input2,input3,input4);
            rrmsg.reply_to = PID::forward(&reply, true);

            if ( m_actor_info->is_local() )
            {
                local_msg<msg_type> *msg = m_actor_info->_allocate_local<msg_type>(typeid(msg_type), m_trace_msgs);
                msg->data = rrmsg;

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) free(msg);
            }
            else
            {
                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(msg_type).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(rrmsg, *serializer) && serializer->flush();
            }

            if (result)
                output = Concurrency::receive<RepT>((Concurrency::ISource<RepT> *)&reply);

            rrmsg.reply_to.invalidate();

            return result;
        }

        /// <summary>
        /// Request is used for duplex messages, where a reply is correlated
        /// with each request message. This is essentially a send/receive
        /// combo, and is a blocking construct.
        /// </summary>
        /// <typeparam name="ReqT">
        /// The outgoing payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The returning payload type 
        /// </typeparam>
        /// <param name="group">A group of actor identifiers</param>
        /// <param name="input">
        /// A reference to the outgoing data.
        /// </param>
        /// <param name="output">
        /// A reference to a vector where the results will be placed.
        /// </param>
        /// <returns>
        /// The result of the serialization and send operation.
        /// </returns>
        template<typename ReqT, typename RepT>
        static bool request(const PID::Group &group, const ReqT &input, std::vector<RepT> &output)
        {
            output.clear();
            output.resize(group.size());

            for (auto pos = group.begin(); pos != group.end(); pos++) if ( !pos->is_valid() || !pos->m_actor_info->chan ) return false;

			const msl::utilities::SafeInt<size_t> groupSize = group.size();
            Concurrency::overwrite_buffer<RepT> *reply = new Concurrency::overwrite_buffer<RepT>[groupSize];
            PID *reply_to = new PID[groupSize];

            bool *result = new bool[groupSize];
        
            for (int i = 0; i < group.size(); i++)
            {
                result[i] = false;
                reply_to[i] = PID::forward((Concurrency::ITarget<RepT> *)(reply+i), true);

                reqrep_msg<ReqT,RepT> rrmsg;
                rrmsg.data = input;
                rrmsg.reply_to = reply_to[i];

                if ( group[i].m_actor_info->is_local() )
                {
                    local_msg<reqrep_msg<ReqT,RepT>> *msg = group[i].m_actor_info->_allocate_local<reqrep_msg<ReqT,RepT>>(typeid(reqrep_msg<ReqT,RepT>), group[i].m_trace_msgs);
                    msg->data = rrmsg;

                    result[i] = group[i].m_actor_info->chan->dispatch((dispatch_msg *)msg);

                    if (!result[i]) free(msg);
                }
                else
                {
                    size_t length = ::calculate_size(rrmsg), len = length;

                    std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(reqrep_msg<ReqT,RepT>).name(), group[i].m_actor_info, group[i].m_trace_msgs));

                    result = (serializer.get() != nullptr) && ::serialize(rrmsg, *serializer) && serializer->flush();
                }
            }

            for (int i = 0; i < group.size(); i++)
            {
                if (result[i])
                    output[i] = Concurrency::receive<RepT>((Concurrency::ISource<RepT> *)(reply+i));

                reply_to[i].invalidate();
            }

            return true;
        }

        /// <summary>
        /// request_forward is used for duplex messages when a reply is correlated
        /// with each request message and there's a PPL message block where replies are
        /// to be delivered.
        /// </summary>
        /// <typeparam name="ReqT">
        /// The outgoing payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The returning payload type 
        /// </typeparam>
        /// <param name="input">
        /// A reference to the outgoing data.
        /// </param>
        /// <param name="output">
        /// A target message block for returning responses into.
        /// </param>
        /// <returns>
        /// The result of the serialization and send operation.
        /// </returns>
        template<typename ReqT, typename RepT>
        bool request_forward(const ReqT &input, Concurrency::ITarget<RepT> &output) const
        {
            if ( !is_valid() ) return false;

            bool result = false;
        
            reqrep_msg<ReqT,RepT> rrmsg;
            rrmsg.data = input;
            rrmsg.reply_to = PID::forward(&output, true);

            if ( m_actor_info->is_local() )
            {
                local_msg<reqrep_msg<ReqT,RepT>> *msg = m_actor_info->_allocate_local<reqrep_msg<ReqT,RepT>>(typeid(reqrep_msg<ReqT,RepT>), m_trace_msgs);
                msg->data = rrmsg;

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) free(msg);
            }
            else
            {
                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(reqrep_msg<ReqT,RepT>).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(rrmsg, *serializer) && serializer->flush();
            }

            return result;
        }

        /// <summary>
        /// request_forward is used for duplex messages when a reply is correlated
        /// with each request message and there's an existing actor identifier where
        /// replies are to be delivered.
        /// </summary>
        /// <typeparam name="ReqT">
        /// The outgoing payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The returning payload type 
        /// </typeparam>
        /// <param name="input">
        /// A reference to the outgoing data.
        /// </param>
        /// <param name="output">
        /// A target actor for sending responses to.
        /// </param>
        /// <returns>
        /// The result of the serialization and send operation.
        /// </returns>
        template<typename ReqT, typename RepT>
        bool request_forward(const ReqT &input, PID output) const
        {
            if ( !is_valid() || !output.is_valid() || !m_actor_info->chan || !output.m_actor_info->chan ) return false;

            bool result = false;
        
            reqrep_msg<ReqT,RepT> rrmsg;
            rrmsg.data = input;
            rrmsg.reply_to = output;

            if ( m_actor_info->is_local() )
            {
                local_msg<reqrep_msg<ReqT,RepT>> *msg = m_actor_info->_allocate_local<reqrep_msg<ReqT,RepT>>(typeid(reqrep_msg<ReqT,RepT>), m_trace_msgs);
                msg->data = rrmsg;

                result = m_actor_info->chan->dispatch((dispatch_msg *)msg);

                if (!result) free(msg);
            }
            else
            {
                std::auto_ptr<Serializer> serializer(_actor_info::_get_msg_serializer(CMD_MESSAGE, typeid(reqrep_msg<ReqT,RepT>).name(), m_actor_info, m_trace_msgs));

                result = (serializer.get() != nullptr) && ::serialize(rrmsg, *serializer) && serializer->flush();
            }
            return result;
        }

    #pragma endregion

    #pragma region forward()

    #pragma region support routines
    private:
        template<typename Payload, typename Functor>
        static IMessageDispatcher _disp_func(Functor &func) 
        {
            return CreateForwarder(false, [=](general_msg *msg) -> bool
            { 
                if ( msg == nullptr ) return false;

                Payload data;
                size_t len = 0;
        
                bool result = false;

                bool same_type = msg->Matches(typeid(Payload).name());
        
                if ( same_type && _actor_info::process_message(msg, data) )
                {
                    try
                    {
                        func(data);
                        result = true;
                    }
                    catch(...)
                    {
                    }
                }
                delete msg;
                return result;
            });
        }

        template<typename Payload>
        static IMessageDispatcher _disp_target(Concurrency::ITarget<Payload> *target)
        {
            return CreateForwarder(false, target);
        }

    public:
    #pragma endregion
        /// <summary>
        /// Forward all messages of type 'Payload' to the specified functor.
        /// </summary>
        /// <typeparam name="Payload">
        /// The payload type 
        /// </typeparam>
        /// <param name="func">The functor that should receive messages of type 'Payload'</param>
        /// <param name="once">If true, only deliver a single </param>
        template<typename Payload, typename Functor>
        static PID forward(Functor &func, bool once = false)
        {
            actor_info result = actor_info(new _actor_info(tcp_listener::get_default()));

            result->chan = _disp_func<Payload, Functor>(func);
    
            // TODO: who removes the registration? TFS: # 336224
            tcp_listener::_register_actor(result->guid, once, result);

            return PID(result);
        }

        /// <summary>
        /// Forward all messages of type 'Payload' to the specified ITarget instance.
        /// </summary>
        /// <typeparam name="Payload">
        /// The payload type 
        /// </typeparam>
        /// <param name="target">The target that should receive messages of type 'Payload'</param>
        /// <param name="once">If true, only deliver a single </param>
        template<typename Payload>
        static PID forward(Concurrency::ITarget<Payload> *target, bool once = false)
        {
            actor_info result = _get_default_info();

            result->chan = _disp_target<Payload>(target);
    
            // TODO: who removes the registration? TFS: # 336224
            _register(result->guid, once, result);

            return PID(result);
        }
    #pragma endregion

    #pragma region Port Creation
        /// <summary>
        /// create_port will create a target block which can be combined with other
        /// PPL data-flow constructs as a pipeline endpoint. It will result in a
        /// message of the 'Payload' type being sent to the actor identified by
        /// the PID that create_port is invoked on.       
        /// </summary>
        /// <typeparam name="Payload">
        /// The payload type 
        /// </typeparam>
        template<typename Payload>
        Concurrency::ITarget<Payload> *create_port() const
        {
            if ( !is_valid() || !m_actor_info->chan ) return nullptr;

            return new Concurrency::call<Payload>([this](Payload data)
            {
                this->send(data);
            });
        }

        /// <summary>
        /// create_request_port will create a target block which can be combined with other
        /// PPL data-flow constructs as a pipeline block. It will result in a
        /// message of the 'ReqT' type being sent to the actor identified by
        /// the PID that create_port is invoked on. As its argument, it takes a pointer
        /// to a target block where reply data is deposited.
        /// </summary>
        /// <typeparam name="ReqT">
        /// The outbound payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The inbound payload type 
        /// </typeparam>
        /// <param name="output">A pointer to a target block where the reply will be deposited.</param>
        template<typename ReqT, typename RepT>
        Concurrency::ITarget<ReqT> *create_request_port(Concurrency::ITarget<RepT> *output) const
        {
            if ( output == nullptr || !is_valid() || !m_actor_info->chan ) return nullptr;

            return new Concurrency::call<ReqT>([this,output](ReqT data)
            {
                this->request_forward(data,*output);
            });
        }

        /// <summary>
        /// create_request_port will create a target block which can be combined with other
        /// PPL data-flow constructs as a pipeline block. It will result in a
        /// message of the 'ReqT' type being sent to the actor identified by
        /// the PID that create_port is invoked on. As its argument, it takes a
        /// identifier for the actor where replies are to be sent.
        /// </summary>
        /// <typeparam name="ReqT">
        /// The outbound payload type 
        /// </typeparam>
        /// <typeparam name="RepT">
        /// The inbound payload type 
        /// </typeparam>
        /// <param name="output">An identifier for the actor where the reply will be sent.</param
        template<typename ReqT, typename RepT>
        Concurrency::ITarget<ReqT> *create_request_port(PID output) const
        {
            if ( !is_valid() || !output.is_valid() || !m_actor_info->chan ) return nullptr;

            return new Concurrency::call<ReqT>([this,output](ReqT data)
            {
                this->request_forward<ReqT,RepT>(data,output);
            });
        }
    #pragma endregion

        _DISPIMP bool migrate(const std::string &to);

    #pragma region private declarations
    private:
        _DISPIMP PID (tcp_listener *);
        _DISPIMP PID (actor_info info) : m_actor_info(info), m_trace_msgs(false) { }

        _DISPIMP std::string to_string() const;

        actor_info m_actor_info;

        bool m_trace_msgs;

        friend class actors::actor;
        friend class actors::_actor_info;
        friend class actors::tcp_listener;
        friend class actors::tcp_proxy;

        // these are wrappers for tcp_listener functions, which prevent cyclic reference in
        // the headers
        _DISPIMP static void _register(const msg_guid &id, bool once_only, actor_info m_actor_info);
        _DISPIMP static actor_info _get_default_info();

        _DISPIMP friend bool (::deserialize)(actors::PID &pid, actors::Serializer &serializer);
        //_DISPIMP friend bool (::deserialize)(actors::PID &pid, actor *proc, actors::Serializer &serializer);
        _DISPIMP friend bool (::serialize)(const actors::PID pid, actors::Serializer &serializer);
        _DISPIMP friend size_t (::calculate_size)(const actors::PID pid);
    #pragma endregion
    };

    /// <summary>
    /// A low-level message definition for messages that carry a PID and a resource string.
    /// </summary>
    /// <remarks>This needs to be defined here because we need the size of PID to be known</remarks>
    struct pid_msg : general_msg
    {
        PID  link_to;           // The PID to link after spawning.
    };

    /// <summary>
    /// A message envelope to hold an identifier to send results to, along with the request data.
    /// </summary>
    template<typename RepT, typename ReqT>
    struct reqrep_msg
    {
        PID  reply_to;
        RepT data;
    };

    /// <summary>
    /// Envelope for faults
    /// </summary>
    struct _fault_msg
    {
        _fault_msg() { }

        _fault_msg(exit_reason reason, PID &faulted) : Reason(reason), Faulted(faulted) { }
        _fault_msg(exit_reason reason, PID &faulted, storage_block &data) : Reason(reason), Faulted(faulted), Data(data) { }

        _fault_msg(const _fault_msg &other) : Reason(other.Reason), Faulted(other.Faulted), Data(other.Data) { }
        void operator = (const _fault_msg &other) { Reason = other.Reason; Faulted = other.Faulted; Data = other.Data; }

        exit_reason  Reason;
        PID          Faulted;
        storage_block Data;
    };
    typedef actors::msg_ptr<_fault_msg> fault_msg;

} // namespace actors

_DISPIMP size_t calculate_size(const actors::pid_msg &msg);
_DISPIMP bool serialize(const actors::pid_msg &msg, actors::Serializer &serializer);
_DISPIMP bool deserialize(actors::pid_msg &msg, actors::Serializer &serializer);

_DISPIMP size_t calculate_size(const actors::fault_msg &msg);
_DISPIMP bool serialize(const actors::fault_msg &msg, actors::Serializer &serializer);
_DISPIMP bool deserialize(actors::fault_msg &msg, actors::Serializer &serializer);

